package syR

import (
	"bufio"
	"db2s/Fp"
	"db2s/Meta"
	"db2s/Store"
	"db2s/arg"
	cp "db2s/componentInit"
	"db2s/full"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/parDef"
	"db2s/pubM"
	quit "db2s/quitDispos"
	"db2s/ref"
	mq "db2s/topic-mq"
	"errors"
	"fmt"
	"github.com/panjf2000/ants/v2"
	"github.com/vbauerster/mpb/v8"
	_ "golang.org/x/net/idna"
	"os"
	"reflect"
	"runtime/pprof"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

var monitorListener mq.MonitorListener

type sendTopicF func(check CheckMode, topic Topic, tableName string, msg map[string]any) bool
type sendCallFuncAction func(s any, d int64)

func GetCheckSchemaMetaData() *CheckSchemaMetaData {
	return &CheckSchemaMetaData{TableObjectFlow: make(chan any, global.ChannelDefault)}
}
func (t CheckSchemaMetaData) ChanSchemaContextObjectInit() *SchemaInitContext {
	var tableFree, tableAccount, breakPointSum int64
	return &SchemaInitContext{ChanSchemaContext: make(chan any, global.ChannelDefault), TableSum: t.ObjectSum, IgnoreObjectSum: t.IgnoreObjectSum, TableAccount: &tableAccount, TableFree: &tableFree, BreakPointSum: &breakPointSum}
}
func breakPointContinue(event string, f any) (r any) {
	//var event = "[breakPointContinue]"
	if f == nil || len(f.([]any)) == 0 {
		return
	}
	_, chanParameter, useParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	v := chanParameter.(arg.InputTableAttributes)
	if _, ok := continueTableList[getTableName(v.Schema, v.Table)]; ok {
		log.MainLog().Info(fmt.Sprintf("%v When the breakpoint resume function is enabled, the current table %v will be skipped.", event, getTableName(v.Schema, v.Table)))
		atomic.AddInt64(useParameter[0].(*int64), 1)
		return true
	}
	return false
}
func schemaMetaExecFuncParameterSplit(f []any) (baseParameter map[string]any, chanParameter any, useParameter []any) {
	baseParameter = f[0].(map[string]any)
	chanParameter = f[1]
	useParameter = f[2:]
	return
}
func getColMe(attributes arg.InputTableAttributes) (result []any, err error) {
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.SrcDBName, GlobalPConfigs.dSns.DestDBName}, "join", attributes.TableAttributes.Role)
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Source.GetDB("single"), Object: parDef.Object{Schema: attributes.Schema, Table: attributes.Table}})
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Target.GetDB("single"), Object: parDef.Object{Schema: attributes.MSchema, Table: attributes.MTable}})
	var columnMe any
	if columnMe, err = c.GetColumnMetaST(); err != nil || columnMe == nil {
		err = ref.ErrAddPrintf("getColMe", err)
		return
	}
	switch columnMe.(type) {
	case []any:
		c.Parameter1[0].Meta = columnMe.([]any)[0]
		c.Parameter1[1].Meta = columnMe.([]any)[1]
	}
	if columnMe, err = c.GetColMetaMapST(); err != nil || columnMe == nil {
		err = ref.ErrAddPrintf("getColMe", err)
		return
	}
	for _, v := range columnMe.([]any) {
		if v != nil {
			result = append(result, v)
		} else {
			result = append(result, v)
		}
	}
	return
}
func getBaseIndexMe(attributes arg.InputTableAttributes) (result any, err error) {
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.SrcDBName}, "left", attributes.TableAttributes.Role)
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Source.GetDB("single"), Object: parDef.Object{Schema: attributes.Schema, Table: attributes.Table}})
	var indexMe any
	if indexMe, err = c.GetNameColumnMetaS(); err != nil {
		err = ref.ErrAddPrintf("getBaseIndexMe", err)
		return
	}
	if indexMe != nil {
		return indexMe, nil
	}
	return
}
func forceMissIndex() (result any, err error) {
	return parDef.IndexColumnMe{
		IndexName: "missIndex",
		ColumnKey: "M",
	}, nil
}
func forceIndexName(attributes arg.InputTableAttributes, im any) (result any, err error) {
	var newForceIndexName = attributes.TableAttributes.ForceIndex
	var newForceIndexCol []string
	if n := strings.Index(attributes.TableAttributes.ForceIndex, "("); n != -1 {
		newForceIndexName = attributes.TableAttributes.ForceIndex[:n]
		pp := attributes.TableAttributes.ForceIndex[n+1:]
		if nn := strings.Index(pp, ")"); nn != -1 {
			newForceIndexCol = strings.Split(pp[:nn], ",")
		}
	}
	for _, v := range im.([]parDef.IndexColumnMe) {
		if strings.EqualFold(newForceIndexName, v.IndexName) {
			var qq []parDef.IdxSubColumnMe
			for _, sv := range v.ColumnMate {
				for _, ssv := range newForceIndexCol {
					if strings.EqualFold(ssv, fmt.Sprintf("%v", sv.ColumnName)) {
						qq = append(qq, sv)
					}
				}
			}
			if len(qq) > 0 {
				v.ColumnMate = qq
			}
			return v, err
		}
	}
	return
}
func forceWhereAdd(attributes arg.InputTableAttributes, im any) (result any, err error) {
	var conditionsCol = make(map[string]int)
	for _, v := range Fp.WhereAddSplit(attributes.TableAttributes.WhereAdd) {
		_, j := Fp.WhereAdd(v)
		for _, l := range j {
			conditionsCol[l["columnName"]]++
		}
	}
	for _, v := range im.([]parDef.IndexColumnMe) {
		var matchBool bool
		for _, sv := range v.ColumnMate {
			if _, ok := conditionsCol[sv.ColumnName]; ok {
				matchBool = true
			}
			if !matchBool {
				break
			}
		}
		if !matchBool {
			continue
		}
		return v, err
	}
	return
}
func forceIndex(attributes arg.InputTableAttributes) (result any, err error) {
	var im any
	if strings.EqualFold(strings.TrimSpace(attributes.TableAttributes.ForceIndex), "missIndex") {
		return forceMissIndex()
	}
	if im, err = getBaseIndexMe(attributes); err != nil {
		err = ref.ErrAddPrintf("forceIndex", err)
		return
	}
	if im == nil {
		return
	}
	switch {
	case len(attributes.TableAttributes.ForceIndex) > 0:
		return forceIndexName(attributes, im)
	case len(attributes.TableAttributes.WhereAdd) > 0:
		return forceWhereAdd(attributes, im)
	}
	return
}
func getIndexMe(attributes arg.InputTableAttributes) (result any, err error) {
	if len(attributes.TableAttributes.ForceIndex) > 0 || len(attributes.TableAttributes.WhereAdd) > 0 {
		return forceIndex(attributes)
	}
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.SrcDBName, GlobalPConfigs.dSns.DestDBName}, "left", attributes.TableAttributes.Role)
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Source.GetDB("single"), Object: parDef.Object{Schema: attributes.Schema, Table: attributes.Table}})
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Target.GetDB("single"), Object: parDef.Object{Schema: attributes.MSchema, Table: attributes.MTable}})
	if result, err = c.GetKeyChooseS(); err != nil || result == nil {
		if err != nil {
			err = ref.ErrAddPrintf("getIndexMe", err)
		}
		return
	}
	return result, err
}
func getPartMe(attributes arg.InputTableAttributes) (result []any, err error) {
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.SrcDBName, GlobalPConfigs.dSns.DestDBName}, "join", attributes.TableAttributes.Role)
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Source.GetDB("single"), Object: parDef.Object{Schema: attributes.Schema, Table: attributes.Table}})
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Target.GetDB("single"), Object: parDef.Object{Schema: attributes.MSchema, Table: attributes.MTable}})
	var partMe any
	if partMe, err = c.GetPartMetaST(); err != nil {
		err = ref.ErrAddPrintf("getPartMe", err)
		return
	}
	if partMe != nil {
		for _, v := range partMe.([]any) {
			if v != nil {
				result = append(result, v)
			} else {
				result = append(result, v)
			}
		}
	}
	return
}
func getObjectMetaFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"struct": reflect.ValueOf(getColMe),
		"index":  reflect.ValueOf(getIndexMe),
		"part":   reflect.ValueOf(getPartMe),
	}
}

func newSchemaInitContextSubOptions(v arg.InputTableAttributes) (res parDef.Options, err error) {
	var pos any
	if pos, err = full.GetGolPosition(full.Gol{
		Drive: []string{GlobalPConfigs.dSns.SrcDBName},
		Join:  "left",
		Parameter1: []parDef.Parameter{
			{Con: GlobalPConfigs.db.Source.GetDB("single"), Options: parDef.Options{Scn: GlobalPConfigs.rules.Sync.Scn}},
		},
	}); err != nil {
		return
	}
	if err = Store.Insert(Store.Position, Store.PositionInfo{
		Key:        fmt.Sprintf("%v_%v", "AA", GlobalPConfigs.dSns.SrcDBName),
		ObjectName: "AA",
		TimeStamp:  time.Now().Format("2006-01-02T15-04-05"),
		Drive:      GlobalPConfigs.dSns.SrcDBName,
		Position:   fmt.Sprintf("%v", pos),
	}); err != nil {
		fmt.Println("--------err:", err)
		return
	}
	return parDef.Options{
		Scn: fmt.Sprintf("%v", pos),
		//StopTime: time.Duration(GlobalPConfigs.rules.Sync.),
		WhereSql:     v.TableAttributes.WhereSql,
		WhereAdd:     Fp.ParseWhereClause(v.TableAttributes.WhereAdd),
		SqlMode:      GlobalPConfigs.rules.Sync.SqlMode,
		TablePx:      v.TableAttributes.PX,
		SqlLogBin:    GlobalPConfigs.rules.Sync.SqlLogBin,
		RowsLimit:    GlobalPConfigs.rules.Sync.RowsLimit,
		RepairMethod: GlobalPConfigs.repair.DataFix,
	}, err
}
func newSchemaInitContext(seq int, v arg.InputTableAttributes, logSeq int64) (result *SchemaInitContext, err error) {
	var options parDef.Options
	if options, err = newSchemaInitContextSubOptions(v); err != nil {
		return
	}
	return &SchemaInitContext{
		Schema: v.Schema, Table: v.Table, MSchema: v.MSchema,
		MTable: v.MTable, Seq: seq, ObjectLogSeq: logSeq,
		Role: v.TableAttributes.Role, TableAttributes: v.TableAttributes,
		Options: options,
	}, err
}
func (t CheckSchemaMetaData) objectMetaInitAbnormalQuit(seq int, v arg.InputTableAttributes, res *SchemaInitContext) bool {
	var err error
	if res.TPod, err = t.podInit(seq, v); err != nil {
		return true
	}
	if atomic.LoadInt64(v.TableStatus) == -1 {
		res.TPod.TableMiss = true
		return true
	}
	return false
}
func reflectValue(v arg.InputTableAttributes) []reflect.Value {
	return []reflect.Value{
		reflect.ValueOf(v),
	}
}
func (t CheckSchemaMetaData) objectMetaResult(logSeq int64, p string, v arg.InputTableAttributes, in []reflect.Value, res *SchemaInitContext) (err error) {
	var event = "[objectMetaInit]"
	for y, x := range in {
		switch y {
		case 0:
			switch p {
			case "struct":
				res.BaseMe.Col = x.Interface()
			case "index":
				res.BaseMe.Index = x.Interface()
			case "part":
				res.BaseMe.Part = x.Interface()
			}
		case 1:
			if x.Interface() != nil {
				err = x.Interface().(error)
				log.ErrorLog().Error(fmt.Sprintf("(%v) %v An error occurred in obtaining the table %v of table %v.%v and will be skipped.error is {%v}", logSeq, event, p, v.Schema, v.Table, err))
				return
			}
		}
	}
	return
}
func (t CheckSchemaMetaData) objectMetaInit(seq int, v arg.InputTableAttributes, logSeq int64) (res *SchemaInitContext, err error) {
	if res, err = newSchemaInitContext(seq, v, logSeq); err != nil {
		return
	}
	if t.objectMetaInitAbnormalQuit(seq, v, res) {
		res.TableMiss = true
		return
	}
	for _, p := range ref.GetJSONFieldNames(res.BaseMe) {
		fun, ok := getObjectMetaFunc()[p]
		if !ok {
			continue
		}
		if err = t.objectMetaResult(logSeq, p, v, fun.Call(reflectValue(v)), res); err != nil {
			err = ref.ErrAddPrintf("objectMetaInit", err)
			return
		}
	}
	return
}

func (t CheckSchemaMetaData) schemaMetaExecFunc(event string, f any) (r any) {
	if f == nil || len(f.([]any)) == 0 {
		return
	}
	baseParameter, chanParameter, userParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	if _, ok := baseParameter["seq"]; !ok {
		return
	}
	if _, ok := baseParameter["logSeq"]; !ok {
		return
	}
	seq := baseParameter["seq"].(int)
	logSeq := baseParameter["logSeq"].(int64)
	if p, err := t.objectMetaInit(seq, chanParameter.(arg.InputTableAttributes), logSeq); err != nil {
		log.ErrorLog().Error(fmt.Sprintf("%v", err))
		return
	} else {
		ChanSchemaContext := userParameter[0].(*SchemaInitContext)
		atomic.AddInt64(ChanSchemaContext.TableAccount, 1)
		atomic.AddInt64(ChanSchemaContext.TableFree, 1)
		ChanSchemaContext.ChanSchemaContext <- p
	}
	return
}
func (t CheckSchemaMetaData) schemaMetaExecQuitFunc(event string, _ any) (r any) {
	ref.MyWaitDel("metaCheck")
	return
}
func (t CheckSchemaMetaData) schemaMetaStartMsgInit(ChanSchemaContext *SchemaInitContext) *mq.MonitorMsgMq {
	var wg sync.WaitGroup
	ref.MyWaitAdd("metaCheck")
	t.schemaMetaMsgSendBody = mq.NewMonitorMsgMq()
	t.schemaMetaMsgSendBody.SendMsg.MsgModifyKey = t.TableObjectFlow
	t.schemaMetaMsgSendBody.NormalQuit = global.EFunc{FuncName: t.schemaMetaExecQuitFunc}
	t.schemaMetaMsgSendBody.ContinueFunc = global.EFunc{FuncName: breakPointContinue, Params: []any{ChanSchemaContext.BreakPointSum}}
	t.schemaMetaMsgSendBody.ExecFunc = append(t.schemaMetaMsgSendBody.ExecFunc, global.EFunc{FuncName: t.schemaMetaExecFunc, Params: []any{ChanSchemaContext}})
	t.schemaMetaMsgSendBody.CurryLimit = mq.SendCurryLimit{
		Wg:       &wg,
		Switch:   true,
		CurrySum: make(chan struct{}, GlobalPConfigs.rules.Mtc),
	}
	return t.schemaMetaMsgSendBody
}
func (t CheckSchemaMetaData) Check() {
	continueTableList = breakP.GetRes("result").(map[string]int)
	ChanSchemaContext := t.ChanSchemaContextObjectInit()
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin init per table object group metadata -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	go func() {
		defer func() {
			close(ChanSchemaContext.ChanSchemaContext)
		}()
		if err := pubM.ObjectMetaDataInit(pubM.ObjectMetaDataPoint{
			NewBreakPoint: pubM.NewBreakPoint{
				BreakPointSum:     ChanSchemaContext.BreakPointSum,
				ContinueTableList: continueTableList,
				ObjectFlow:        t.TableObjectFlow,
				TableAccount:      ChanSchemaContext.TableAccount,
				TableFree:         ChanSchemaContext.TableFree,
			},
			ExecFunc: global.EFunc{FuncName: t.schemaMetaExecFunc, Params: []any{ChanSchemaContext}},
		}).NormalChanSendMsg("syRCheck"); err != nil {
			return
		}
	}()
	global.GetIoc().RegisterContext("schemaInitContext", ChanSchemaContext)
}
func FuncExportVar() {
	//函数调用映射
	FuncRef = global.GetIoc().GetBean("checkObjectFuncContext").(map[string]reflect.Value)
	ScheduleTaskInitFuncRef = global.GetIoc().GetBean("scheduleTaskInitFuncContext").(map[string]reflect.Value)
	closeObject = global.GetIoc().GetBean("closeObject").(quit.Close)
	MqListenerSInitFunc = global.GetIoc().GetBean("mqListenerSInitFuncContext").(map[string]reflect.Value)
}

// ExportVar 导出变量 方便使用
func ExportVar() {
	//producer 生产者
	producer = global.GetIoc().GetBean("producer").(mq.Producer)
	//consumer 消费者
	consumer = global.GetIoc().GetBean("consumer").(mq.Consumer)
	//monitorListener mq 监控
	monitorListener = global.GetIoc().GetBean("monitorListener").(mq.MonitorListener)
	PlanContext = global.GetIoc().GetBean("schedulePlanContext").(*SchedulePlanContext) //注册ioc 容器
}

func (context *SchedulePlanContext) GetIndexMode() CheckMode {
	switch global.GetIoc().GetBean("globalConfig").(*GlobalParametersConfig).rules.CheckMode {
	case "sync":
		return "sync"
	}
	return "sync"
}
func getCheckMod() CheckMode {
	switch global.GetIoc().GetBean("globalConfig").(*GlobalParametersConfig).rules.CheckMode {
	case "sync":
		return PlanContext.GetIndexMode()
	}
	return ""
}

func (context *SchedulePlanContext) objectEndMark(index CheckMode) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].ObjectEndMark()
}
func syncToTopic(plan *SchedulePlan2) {
	sendMsg(getCheckMod(), syncIsBaseQuery, getTableName(plan.Object.Schema, plan.Object.Table), map[string]any{"segmentSql": map[string]string{}, "plan": plan, "subTask": plan.subTaskInfo, "topic": string(syncIsBaseQuery)})
}
func checkObjectFunc() map[string]reflect.Value {
	return map[string]reflect.Value{"sync": reflect.ValueOf(syncToTopic)}
}
func sendScheduleMsgFunc() map[string]reflect.Value {
	return map[string]reflect.Value{"sync": reflect.ValueOf(initSyncSchedulePlan)}
}
func mqListenerSInitFunc() map[string]reflect.Value {
	return map[string]reflect.Value{"sync": reflect.ValueOf(syncDataMQListenInitFunc)}
}
func schedulePlanGetIocInit() {
	global.GetIoc().RegisterContext("checkObjectFuncContext", checkObjectFunc())
	global.GetIoc().RegisterContext("scheduleTaskInitFuncContext", sendScheduleMsgFunc())
	global.GetIoc().RegisterContext("mqListenerSInitFuncContext", mqListenerSInitFunc())
	FuncExportVar()
}
func schedulePlanThreadPoolInit() (pool *ants.Pool, err error) {
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	//初始化上下文线程池
	if pool, err = ants.NewPool(initContext.TableSum); err != nil {
		return
	}
	progress := mpb.New(mpb.WithWidth(80))
	global.GetIoc().RegisterContext("progress", progress)
	return
}
func planContextBarInit(config *SchedulePlanContextConfig) *mq.TaskBar {
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	return cp.InitBarMonitor(cp.Bar{
		TaskTotal:     initContext.TableAccount,
		Task:          initContext.TableSum,
		Free:          initContext.TableFree,
		BreakPointSum: initContext.BreakPointSum,
		CheckMod:      config.GtCheckConfig.SecondaryL.RulesV.CheckMode,
		Object:        config.GtCheckConfig.SecondaryL.SchemaV.CheckIndexObject,
		BeginTime:     config.BeginTime,
		TtyBarFin:     GlobalPConfigs.FilePoint.TtyBarFin,
	}, len(GetCheckToTopicMQListeners()[cp.CheckMode(config.GtCheckConfig.SecondaryL.RulesV.CheckMode)]))
}
func PlanContextInit(config *SchedulePlanContextConfig) (PlanContext *SchedulePlanContext, err error) {
	var pool *ants.Pool
	//初始化上下文
	if pool, err = schedulePlanThreadPoolInit(); err != nil {
		return
	}
	PlanContext = &SchedulePlanContext{
		Pool:                      pool,
		AutoCloseMqComponentTable: cp.InitAutoCloseMqComponentTable(),
		BeginTime:                 config.BeginTime,
	}
	PlanContext.TaskBarTable = planContextBarInit(config) //初始化执行计划表
	PlanContext.ChanSchedulePlanTable = InitToSchedulePlan2Table()
	global.GetIoc().RegisterContext("schedulePlanContext", PlanContext) //注册ioc 容器
	return
}

// NewSchedulePlanContext 创建执行计划上下文
func schedulePlanContext(config *SchedulePlanContextConfig) (PlanContext1 *SchedulePlanContext) {
	var err error
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin init schedule task Connext -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	lock.Lock()
	defer lock.Unlock()
	schedulePlanGetIocInit()
	if err = initContextBean(config); err != nil {
		fmt.Println("创建执行计划上下文 NewSchedulePlanContext 失败", err)
		return nil
	}
	if PlanContext1, err = PlanContextInit(config); err != nil {
		return nil
	}
	//导出变量
	ExportVar()
	return
}

// initContextBean 初始化上下文需要用到的组件
func initContextBean(config *SchedulePlanContextConfig) error {
	//TODO: checkConfig 行哥 你来写
	createMq := config.MQConfig.CreateMq(GetCheckToTopicMQListeners())
	if createMq == nil {
		return errors.New("初始化 mq 失败")
	}
	return nil
}

func (context *SchedulePlanContext) Close() {
	for _, component := range context.AutoCloseMqComponentTable {
		context.wg.Add(1)
		func() {
			defer context.wg.Done()
			component.Close(false)
		}()
	}
	context.wg.Wait()
}

func missTableAction(k *SchedulePlan2) {
	event := "[miss]"
	log.MainLog().Info(fmt.Sprintf("(%v) %v Table %v.%v does not exist on the target side, and the table status is {missTable}", k.Object.ObjectLogSeq, event, k.Object.Schema, k.Object.Table))
	newTer.Write(k.TPod)
	k.XlsMiss.Data.Missing = GlobalPConfigs.dSns.DestDBName
	newCell.Write(*k.XlsMiss)
	PlanContext.TaskBarSubsStatusSwap(k.subTaskInfo, "finish")
}
func sendStartMsgFinishFunc(event string, _ any) (r any) {
	ref.MyWaitDel("schedulePlan")
	ref.MyWaitClose()
	PlanContext.objectEndMark(getCheckMod())
	return
}
func tableObjectErrorStatus(k *SchedulePlan2) bool {
	if atomic.LoadInt64(k.Status.ErrEnd) >= 0 {
		return false
	}
	newTer.Write(k.TPod)
	newCell.Write(k.XlsTerminal)
	newCell.Write(k.XlsResult)
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(k.subTaskInfo, "error")
	}

	return true
}
func sendStartMsgContinueFunc(event string, f any) (r any) {
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	k := chanParameter.(*SchedulePlan2)
	if k.TPod.TableMiss {
		missTableAction(k)
		return true
	}
	return tableObjectErrorStatus(k)
}
func sendStartMsgExecFunc(event string, f any) (r any) {
	log.SendMsg().Debug(event, fmt.Sprintf("sendStartMsgExecFunc start run exec Func. parameter is %v.", f))
	_, chanParameter, _ := schemaMetaExecFuncParameterSplit(f.([]any))
	if funcName, ok := FuncRef[GlobalPConfigs.rules.CheckMode]; ok {
		if funcName.Kind() == reflect.Func {
			arguments := []reflect.Value{reflect.ValueOf(chanParameter)}
			funcName.Call(arguments)
		}
	} else {
		fmt.Println(fmt.Sprintf("-- [%v] %v report: checkObject parameter selection error, please refer to the log file for details, enable debug to get more information -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		os.Exit(1)
	}
	return
}
func sendStartMsgInit() *mq.MonitorMsgMq {
	var wg sync.WaitGroup
	ref.MyWaitAdd("schedulePlan")
	baseMsgSendBody = mq.NewMonitorMsgMq()
	baseMsgSendBody.SendMsg.MsgModifyKey = PlanContext.ChanSchedulePlanTable
	baseMsgSendBody.NormalQuit = global.EFunc{FuncName: sendStartMsgFinishFunc}
	baseMsgSendBody.AbnormalQuit = global.EFunc{FuncName: sendStartMsgFinishFunc}
	baseMsgSendBody.ContinueFunc = global.EFunc{FuncName: sendStartMsgContinueFunc}
	baseMsgSendBody.ExecFunc = append(baseMsgSendBody.ExecFunc, global.EFunc{FuncName: sendStartMsgExecFunc})
	baseMsgSendBody.CurryLimit = mq.SendCurryLimit{
		Switch:     true,
		Wg:         &wg,
		ManualDone: true,
		CurrySum:   make(chan struct{}, GlobalPConfigs.rules.Mtc),
	}
	return baseMsgSendBody
}
func schedulePlanStart1() {
	if PlanContext.Pool.Submit(func() {
		//发送消息
		fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task sync -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		sendStartMsgInit().NormalChanSendMsg("syR -> func(schedulePlanStart1)")
		fmt.Println(fmt.Sprintf("-- [%v] %v report: schedule task sync finish -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	}) != nil {
		return
	}
}
func (context *SchedulePlanContext) monitor() {
	switch GlobalPConfigs.result.Teletypewriter {
	case "bar":
		context.TaskBarTable.BarMonitor()
	case "topic":
		monitorListener.TurnMonitor()
	default:
		context.TaskBarTable.BarMonitor()
	}
}
func (context *SchedulePlanContext) closeMonitor() {
	context.Close()
	//关闭消费者 阻塞
	consumer.Shutdown()
	//关闭生产者
	producer.Shutdown()
	newTer.Stop()
	newCell.Stop()
	context.TaskBarTable.BarCloseAction()
}
func (context *SchedulePlanContext) endClose() {
	if err := xlsFile.Close(); err != nil {
		return
	}
	closeObject.CloseAction()
	fmt.Println("")
	//输出结果信息
	fmt.Println("Check end")
	log.MainLog().Info(fmt.Sprintf("%v check object {%v} complete !!!", arg.ToolName, GlobalPConfigs.rules.CheckMode))
	fmt.Println(fmt.Sprintf("** %v Overview of results **", arg.ToolName))
	fmt.Println("Check time: ", fmt.Sprintf("%.2fs", time.Since(context.BeginTime).Seconds()), "(Seconds)")
	if err := newTer.Load("execTime", fmt.Sprintf("%v", time.Since(context.BeginTime).Seconds())); err != nil {
		log.ErrorLog().Error(fmt.Sprintf("%v", err))
	}
	newTer.Print()
}
func (context *SchedulePlanContext) Start(wg *sync.WaitGroup) {
	fmt.Println(fmt.Sprintf("-- [%v] %v report: start schedule task monitor -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	terminal(wg)
	context.monitor()
	Cell(wg)
}
func (context *SchedulePlanContext) Check() {
	var wg sync.WaitGroup
	context.Start(&wg)
	closeObject.AbnormalExit()
	//执行计划开始
	schedulePlanStart1()
	ref.MyWaitWait()
	context.Close()
	context.closeMonitor()
	//关闭检测
	wg.Wait()
	context.endClose()
}
func truncateInteraction(input string) {
	// 检查是否输入 'exit'
	switch strings.TrimSpace(strings.ToLower(input)) {
	case "yes":
		fmt.Println(fmt.Sprintf("[%v] %v report: Start data migration operation.", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		return
	case "no":
		fmt.Println(fmt.Sprintf("[%v] %v report: The program starts to exit.", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		os.Exit(1)
	default:
		fmt.Println(fmt.Sprintf("[%v] %v report: Please enter correct parameters.", arg.ToolName, time.Now().Format("2006-01-02 15:04:05")))
	}
}

/*
当输入参数truncate为true时，则进行交互确认
*/
func interaction() {
	scanner := bufio.NewScanner(os.Stdin)
	for {
		// 提示用户输入
		fmt.Print(fmt.Sprintf("[%v] %v report: warn! This operation will perform a truncate table operation! ! ! \nPlease enter \"yes\" to continue or \"no\" to exit: ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		// 读取输入
		if scanner.Scan() {
			truncateInteraction(scanner.Text())
		}
		// 检查扫描器是否出错
		if err := scanner.Err(); err != nil {
			fmt.Println(fmt.Sprintf("[%v] %v report: Input parameter error!! Please enter correct parameters.", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
		}
	}
}

// Audit input
func Audit(m arg.ConfigParameter) {
	fmt.Println(fmt.Sprintf("-- [%v] %v report: begin input sync mode -- ", time.Now().Format("2006-01-02 15:04:05"), arg.ToolName))
	defer recoverException("Audit")
	if m.ArgInput.RowsDataCmd.Sync.Optimizer.CpuOptimizer {
		defer pprof.StopCPUProfile()
		ref.CpuAnalyze()
	}
	if m.ArgInput.RowsDataCmd.Sync.Optimizer.MemOptimizer {
		f := ref.MemAnalyze()
		defer func() {
			if err := f.Close(); err != nil {
				return
			}
		}()
	}
	if m.SecondaryL.RulesV.Sync.Truncate && !m.ArgInput.RowsDataCmd.Sync.Truncate {
		interaction()
	}
	ref.MyWaitInit()
	schemaContext(&SchemaInitContextConfig{
		GtCheckConfig: m,
	}).Check()
	schedulePlanContext(&SchedulePlanContextConfig{
		GtCheckConfig: m,
		MQConfig:      cp.NewDefaultMq(),
		BeginTime:     m.BeginTime,
	}).Check()
}
